package streamproc.buffercomponents;

import java.sql.Timestamp;
import java.util.Collections;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Vector;
import streamdef.StreamCondition;
import streamdef.StreamDefinitionManager;
import streamdef.ConditionStatement;
import streamdef.ConditionStatementFieldFromOneStream;
import streamdef.ConditionStatementFieldFromTwoStreams;
import streamdef.SubCondition;
import streamdef.DEFINES.FieldRelationType;
import streamdef.DEFINES.ConditionStatementType;
import event.Event;
import event.EventFormatManager;
import event.DEFINES.CEPLEventDataFieldType;
import event.UnknownDataFieldException;

public class Buffer {
	private static double defaultRange = 3600.0; // 1 hour
	private long streamID = -1;
	private double range = 0.0;
	private ArrayList<Event> sortedEventsByTimestamp = null;

	// multiple index, besides the timestamp
	// if multiple events have the same value for the index
	// -> the most recent one is stored
	// (numbers are Doubles, text indices are Strings)
	private HashMap<String, HashMap<Object, Vector<Event>>> multipleIndexingHashMaps = null;
	private Vector<String> multipleIndexingUsedFields = null;

	public Buffer(long streamID) {
		sortedEventsByTimestamp = new ArrayList<Event>();
		multipleIndexingHashMaps = new HashMap<String, HashMap<Object, Vector<Event>>>();
		multipleIndexingUsedFields = new Vector<String>();
		this.streamID = streamID;
		setTotalRange();
		setHashMapsForMultipleIndexing();
	}

	public void reset() {
		for (Object key : multipleIndexingHashMaps.keySet()) {
			HashMap<Object, Vector<Event>> h = multipleIndexingHashMaps
					.get(key);
			for (Object key2 : h.keySet()) {
				Vector<Event> v = h.get(key2);
				v.clear();
			}
		}

		sortedEventsByTimestamp.clear();
		multipleIndexingHashMaps.clear();
		multipleIndexingUsedFields.clear();
	}

	protected void finalize() {
		reset();
	}

	public int getNbOfStoredEvents() {
		return sortedEventsByTimestamp.size();
	}

	// seconds
	public double getTotalRange() {
		return range;
	}

	private void setTotalRange() {
		range = 0.0;

		Vector<Long> streamIDs = StreamDefinitionManager.instance()
				.getStreamIDs();
		for (int i = 0; i < streamIDs.size(); ++i) {
			StreamCondition mainQ = StreamDefinitionManager.instance()
					.getParsedQuery(streamIDs.get(i));
			Vector<Long> internalIDs = mainQ
					.getInternalStreamIDsForExternalStreamID(streamID);
			if (internalIDs.size() == 0)
				;
			else {
				for (int j = 0; j < internalIDs.size(); ++j) {
					// A buffer is only needed for streams which may not occur
					// in certain queries -> only use those ranges
					if (!mainQ.mayOccur(internalIDs.get(j))) {
						double r = mainQ
								.getRangeForInternalStreamID(internalIDs.get(j));
						if (r == -1)
							r = defaultRange;
						if (r > range)
							range = r;
					}
				}
			}
		}
	}

	private void setHashMapsForMultipleIndexing() {
		StreamDefinitionManager pqm = StreamDefinitionManager.instance();
		Vector<Long> otherIDs = pqm.getStreamIDs();
		for (int i = 0; i < otherIDs.size(); ++i) {
			long ID = otherIDs.get(i);
			if (ID != this.streamID) {
				StreamCondition mq = pqm.getParsedQuery(ID);
				Vector<Long> forbiddenInternalIDs = mq
						.getNonOccuringInternalStreamIDs();
				Vector<Long> usedInternalIDs = new Vector<Long>();
				boolean usedInMQ = false;
				for (int j = 0; j < forbiddenInternalIDs.size(); ++j) {
					long forbiddenExternalID = mq
							.getExternalStreamID(forbiddenInternalIDs.get(j));
					if (forbiddenExternalID == this.streamID) {
						usedInMQ = true;
						usedInternalIDs.add(forbiddenInternalIDs.get(j));
					}
				}
				if (usedInMQ) {
					Vector<SubCondition> subConditions = mq.getSubQueries();
					for (int j = 0; j < subConditions.size(); ++j) {
						SubCondition sq = subConditions.get(j);
						boolean usedInSQ = false;
						for (int k = 0; k < usedInternalIDs.size(); ++k)
							if (sq.getUsedInternalStreamIDs().contains(
									usedInternalIDs.get(k)))
								usedInSQ = true;

						if (usedInSQ) {
							Vector<ConditionStatement> queryStats = sq
									.getQueryStatements();
							for (int k = 0; k < queryStats.size(); ++k) {
								ConditionStatement qs = queryStats.get(k);
								if (qs.getType() == ConditionStatementType.FIELD_ONESTREAM) {
									ConditionStatementFieldFromOneStream qs_ = (ConditionStatementFieldFromOneStream) qs;
									if (usedInternalIDs
											.contains(qs_.internalStreamID))
										this.multipleIndexingUsedFields
												.add(qs_.field);
								} else if (qs.getType() == ConditionStatementType.FIELD_TWOSTREAMS) {
									ConditionStatementFieldFromTwoStreams qs_ = (ConditionStatementFieldFromTwoStreams) qs;
									if (usedInternalIDs
											.contains(qs_.internalStreamID1))
										this.multipleIndexingUsedFields
												.add(qs_.field1);
									if (usedInternalIDs
											.contains(qs_.internalStreamID2))
										this.multipleIndexingUsedFields
												.add(qs_.field2);
								}
							}
						}
					}
				}
			}
		}

		for (int i = 0; i < multipleIndexingUsedFields.size(); ++i) {
			this.multipleIndexingHashMaps.put(
					multipleIndexingUsedFields.get(i),
					new HashMap<Object, Vector<Event>>());
		}
	}

	public synchronized void processEvent(Event e) {
		addSorted(e);
		// addToMultipleIndexing(e);
	}

	// start at the back of the buffer !!
	private void addSorted(Event e) {
		Timestamp ts = e.getTimestamp();
		int index;
		boolean stop = false;
		for (index = sortedEventsByTimestamp.size() - 1; index >= 0 && !stop; --index) {
			if (sortedEventsByTimestamp.get(index).getTimestamp().before(ts)) {
				stop = true;
				index++;
			} else if (!sortedEventsByTimestamp.get(index).getTimestamp()
					.before(ts)
					&& !sortedEventsByTimestamp.get(index).getTimestamp()
							.after(ts)) {
				stop = true;
				index++;
			}
		}

		sortedEventsByTimestamp.add(index + 1, e);
		e.setBufferIndex(index + 1);
	}

	@SuppressWarnings("unused")
	private void addToMultipleIndexing(Event e) {
		for (int i = 0; i < multipleIndexingUsedFields.size(); ++i) {
			String field = multipleIndexingUsedFields.get(i);
			HashMap<Object, Vector<Event>> hashMap = multipleIndexingHashMaps
					.get(field);

			CEPLEventDataFieldType type = EventFormatManager.instance()
					.getFieldValueType(this.streamID, field);
			if (type != CEPLEventDataFieldType.TEXT
					&& type != CEPLEventDataFieldType.TEXT) {
				// put double as index for numerical values
				double d = e.getEncapsulatingDouble(field);
				Vector<Event> v = hashMap.get(d);
				if (v == null) {
					v = new Vector<Event>();
					v.add(e);
					hashMap.put(d, v);
				} else
					v.add(e);
			} else if (type == CEPLEventDataFieldType.TEXT) {
				try {
					String d = e.getString(field);
					Vector<Event> v = hashMap.get(d);
					if (v == null) {
						v = new Vector<Event>();
						v.add(e);
						hashMap.put(new String(d), v);
					} else
						v.add(e);
				} catch (UnknownDataFieldException e1) {
					e1.printStackTrace();
				}
			}
		}
	}

	public Event getOldestEvent() {
		return sortedEventsByTimestamp.get(0);
	}

	public Event getNewestEvent() {
		return sortedEventsByTimestamp.get(sortedEventsByTimestamp.size() - 1);
	}

	public boolean checkValidnessOrder() {
		for (int i = 0; i < sortedEventsByTimestamp.size() - 1; ++i) {
			if (sortedEventsByTimestamp.get(i).getTimestamp()
					.after(sortedEventsByTimestamp.get(i + 1).getTimestamp()))
				return false;
		}

		return true;
	}

	public synchronized int removeOutdatedEventsBinarySearch(Timestamp current) {
		Timestamp ts = new Timestamp(current.getTime()
				- (long) (range * 1000.0));
		int firstValidIndex = Collections.binarySearch(sortedEventsByTimestamp,
				ts);
		if (firstValidIndex < 0)
			firstValidIndex = -(firstValidIndex + 1);

		// get the subset
		List<Event> eventsToRemove = sortedEventsByTimestamp.subList(0,
				firstValidIndex);
		for (int i = 0; i < eventsToRemove.size(); ++i)
			eventsToRemove.get(i).setRemoveFromBuffer(true);

		// clear the subset itself
		eventsToRemove.subList(0, firstValidIndex).clear();
		sortedEventsByTimestamp.trimToSize();
		for (int i = 0; i < sortedEventsByTimestamp.size(); ++i)
			sortedEventsByTimestamp.get(i).setBufferIndex(i);

		for (Object key : multipleIndexingHashMaps.keySet()) {
			HashMap<Object, Vector<Event>> h = multipleIndexingHashMaps
					.get(key);
			for (Object key2 : h.keySet()) {
				Vector<Event> v = h.get(key2);

				for (int i = 0; i < v.size(); ++i) {
					Event x = v.get(i);
					if (x.removeFromBuffer()) {
						v.remove(i);
						--i;
					}
				}

				if (v.size() == 0) {
					h.remove(key2);
				}
			}
		}

		return firstValidIndex;
	}

	// return = index of valid events in the buffer
	public synchronized Vector<Integer> valueExists(String field,
			FieldRelationType relation, Object value, boolean isDouble,
			Vector<Integer> bufferIndexLimits,
			Vector<Integer> bufferIndicesToCheck, Vector<Long> eventIDsToExclude) {
		Vector<Integer> result = new Vector<Integer>();

		if (bufferIndexLimits.size() != 2)
			return null;
		if (bufferIndicesToCheck != null && bufferIndicesToCheck.size() == 0)
			return null;

		int leftLimit = bufferIndexLimits.get(0);
		int rightLimit = bufferIndexLimits.get(1);

		CEPLEventDataFieldType type = EventFormatManager.instance()
				.getFieldValueType(streamID, field);

		if (type != CEPLEventDataFieldType.TEXT) {
			Double doubleValueToCheckWith = (Double) value;
			if (!isDouble) {
				String s = (String) value;
				doubleValueToCheckWith = Double.parseDouble(s);
			}

			if (bufferIndicesToCheck == null) {
				for (int i = leftLimit; i <= rightLimit; ++i) {
					Event e = sortedEventsByTimestamp.get(i);
					Double d = e.getEncapsulatingDouble(field);

					if (relation == FieldRelationType.EQUALS
							&& d == doubleValueToCheckWith)
						result.add(i);
					else if (relation == FieldRelationType.NOTEQUALS
							&& d != doubleValueToCheckWith)
						result.add(i);
					else if (relation == FieldRelationType.EQUALSORGREATER
							&& d >= doubleValueToCheckWith)
						result.add(i);
					else if (relation == FieldRelationType.EQUALSORLESSER
							&& d <= doubleValueToCheckWith)
						result.add(i);
					else if (relation == FieldRelationType.GREATER
							&& d > doubleValueToCheckWith)
						result.add(i);
					else if (relation == FieldRelationType.LESSER
							&& d < doubleValueToCheckWith)
						result.add(i);
				}
			} else {
				for (int i = 0; i < bufferIndicesToCheck.size(); ++i) {
					Event e = sortedEventsByTimestamp.get(bufferIndicesToCheck
							.get(i));
					Double d = e.getEncapsulatingDouble(field);

					if (relation == FieldRelationType.EQUALS
							&& d == doubleValueToCheckWith)
						result.add(bufferIndicesToCheck.get(i));
					else if (relation == FieldRelationType.NOTEQUALS
							&& d != doubleValueToCheckWith)
						result.add(bufferIndicesToCheck.get(i));
					else if (relation == FieldRelationType.EQUALSORGREATER
							&& d >= doubleValueToCheckWith)
						result.add(bufferIndicesToCheck.get(i));
					else if (relation == FieldRelationType.EQUALSORLESSER
							&& d <= doubleValueToCheckWith)
						result.add(bufferIndicesToCheck.get(i));
					else if (relation == FieldRelationType.GREATER
							&& d > doubleValueToCheckWith)
						result.add(bufferIndicesToCheck.get(i));
					else if (relation == FieldRelationType.LESSER
							&& d < doubleValueToCheckWith)
						result.add(bufferIndicesToCheck.get(i));
				}
			}

			if (result.size() == 0)
				return null;
			else
				return result;

		} else {
			String stringToCheckWith = (String) value;
			if (isDouble)
				stringToCheckWith = value.toString();

			if (bufferIndicesToCheck == null) {
				for (int i = leftLimit; i <= rightLimit; ++i) {
					Event e = sortedEventsByTimestamp.get(i);
					String s = null;
					try {
						s = e.getString(field);
					} catch (UnknownDataFieldException e1) {
						e1.printStackTrace();
					}

					if (relation == FieldRelationType.EQUALS
							&& stringToCheckWith.compareToIgnoreCase(s) == 0)
						result.add(i);
					else if (relation == FieldRelationType.NOTEQUALS
							&& stringToCheckWith.compareToIgnoreCase(s) != 0)
						result.add(i);
				}
			} else {
				for (int i = 0; i < bufferIndicesToCheck.size(); ++i) {
					Event e = sortedEventsByTimestamp.get(bufferIndicesToCheck
							.get(i));
					String s = null;
					try {
						s = e.getString(field);
					} catch (UnknownDataFieldException e1) {
						e1.printStackTrace();
					}

					if (relation == FieldRelationType.EQUALS
							&& stringToCheckWith.compareToIgnoreCase(s) == 0)
						result.add(bufferIndicesToCheck.get(i));
					else if (relation == FieldRelationType.NOTEQUALS
							&& stringToCheckWith.compareToIgnoreCase(s) != 0)
						result.add(bufferIndicesToCheck.get(i));
				}
			}

			if (result.size() == 0)
				return null;
			else
				return result;
		}
	}

	// return = index of interval limits in buffer, inclusive
	// inclusive left and right limit
	public synchronized Vector<Integer> eventExists(Timestamp begin,
			Timestamp end, Vector<Long> eventIDsToExclude) {
		if(streamID == 3)
			;
		
		
		Vector<Integer> result = new Vector<Integer>();

		if (end == null) {
			int indexBegin = Collections.binarySearch(sortedEventsByTimestamp,
					begin);

			if (indexBegin >= sortedEventsByTimestamp.size())
				return null;
			else if (indexBegin >= 0) {
				for (int i = indexBegin; i < sortedEventsByTimestamp.size(); ++i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
						result.add(sortedEventsByTimestamp.size() - 1);
						return result;
					}
				return null;
			} else {
				indexBegin = -(indexBegin + 1);

				for (int i = indexBegin; i < sortedEventsByTimestamp.size(); ++i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
						result.add(sortedEventsByTimestamp.size() - 1);
						return result;
					}
				return null;
			}
		} else if (begin == null) {
			int indexEnd = Collections.binarySearch(sortedEventsByTimestamp,
					end);
			if (indexEnd >= 0) {
				for (int i = indexEnd; i >= 0; --i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(0);
						result.add(i);
						return result;
					}
				return null;
			} else {
				indexEnd = -(indexEnd + 1);

				for (int i = indexEnd - 1; i >= 0; --i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(0);
						result.add(i);
						return result;
					}
				return null;
			}
		} else {
			if (begin.after(end))
				return null;

			int indexBegin = Collections.binarySearch(sortedEventsByTimestamp,
					begin);
			int indexEnd = Collections.binarySearch(sortedEventsByTimestamp,
					end);

			if (indexBegin >= sortedEventsByTimestamp.size())
				return null;
			else if (indexBegin >= 0 && indexEnd >= 0) {
				boolean skip = false;
				for (int i = indexBegin; i <= indexEnd && !skip; ++i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
						skip = true;
					}
				for (int i = indexEnd; i >= indexBegin; --i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
					}

				if (result.size() == 0)
					return null;
				else
					return result;
			} else if (indexBegin < 0 && indexEnd >= 0) {
				boolean skip = false;
				indexBegin = -(indexBegin + 1);

				for (int i = indexBegin; i <= indexEnd && !skip; ++i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
						skip = true;
					}
				for (int i = indexEnd; i >= indexBegin; --i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
					}

				if (result.size() == 0)
					return null;
				else
					return result;
			} else if (indexBegin >= 0 && indexEnd < 0) {
				boolean skip = false;
				indexEnd = -(indexEnd + 1);

				for (int i = indexBegin; i < indexEnd && !skip; ++i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
						skip = true;
					}
				for (int i = indexEnd - 1; i >= indexBegin; --i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
					}

				if (result.size() == 0)
					return null;
				else
					return result;
			} else if (indexBegin < 0 && indexEnd < 0) {
				boolean skip = false;
				indexEnd = -(indexEnd + 1);
				indexBegin = -(indexBegin + 1);

				for (int i = indexBegin; i < indexEnd && !skip; ++i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
						skip = true;
					}
				for (int i = indexEnd - 1; i >= indexBegin; --i)
					if (eventIDsToExclude == null
							|| !eventIDsToExclude
									.contains(sortedEventsByTimestamp.get(i)
											.getID())) {
						result.add(i);
					}

				if (result.size() == 0)
					return null;
				else
					return result;
			}
		}
		return null;
	}
}
